package com.xuan.temper.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xuan.temper.dao.ExponentialTemperatureDao;
import com.xuan.temper.entity.*;
import com.xuan.temper.service.EchartsService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @创建人： king xuan
 * @创建时间：2020/9/16
 * @描述：
 */
@Component
public class MQPushConsumer implements MessageListenerConcurrently {
    private static final Logger LOGGER = LoggerFactory.getLogger(MQPushConsumer.class);

    @Autowired
    private ExponentialTemperatureDao exponentialTemperatureDao;

    @Autowired
    private EchartsService echartsService;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
    // 1. 创建默认消息消费者(指定组名)
    private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("RocketMQPushConsumer");

    /**
     * 初始化
     *
     * @throws MQClientException
     */
    @PostConstruct
    public void start() {
        try {
            LOGGER.info("MQ：启动消费者");

            consumer.setNamesrvAddr(namesrvAddr);

            // 1.2 设置从哪里开始消费
            // ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET:从队列头部开始消费
            // ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET:从队列尾部开始消费
            // 从消息队列头开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            // 订阅主题
            consumer.subscribe("data", "*");
            // 注册消息监听器
            consumer.registerMessageListener(this);
            // 启动消费端
            consumer.start();
        } catch (MQClientException e) {
            LOGGER.error("MQ：启动消费者失败：{}-{}", e.getResponseCode(), e.getErrorMessage());
            throw new RuntimeException(e.getMessage(), e);
        }

    }

    /**
     * 消费消息
     *
     * @param msgs
     * @param context
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        int index = 0;
        try {
            for (; index < msgs.size(); index++) {
                MessageExt msg = msgs.get(index);

                String messageBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);

               // LOGGER.info("MQ：消费者接收新信息 {} {} {} {} {}", msg.getMsgId(), msg.getTopic(), msg.getTags(), msg.getKeys(), messageBody);
                LOGGER.info("MQ：消费者接收新信息 {} {} {} {} {}");

            Data jsonValue = JSON.parseObject(messageBody, Data.class);
            ExponentialTemperature temperature = new ExponentialTemperature();
            temperature.setPeJq(jsonValue.getPe_ttm().getMcw());//pe值
            temperature.setPeTemp(jsonValue.getPe_ttm().getPeTemper());//pe温度
            temperature.setPbJq( jsonValue.getPb().getMcw());//pb值
            temperature.setPbTemp(jsonValue.getPb().getPbTemper());//pb温度
            temperature.setCurrenTemp(jsonValue.getCurrenTemp());//指数温度
            temperature.setStockCode(jsonValue.getStockCode());//指数代码
            temperature.setName(jsonValue.getStockName());//指数名称
            temperature.setDyr(jsonValue.getDyr().getMcw());//股息率
            temperature.setCp(jsonValue.getCp());//市值
            temperature.setCpc(jsonValue.getCpc());//涨跌幅
            temperature.setTime(jsonValue.getDate());
            temperature.setUpdateTime(new Date());
            SimpleDateFormat sg =new SimpleDateFormat("yyyy-MM-dd");
            String format = sg.format(jsonValue.getDate());
            System.out.println(format);
            Boolean delete = exponentialTemperatureDao.delete(sg.format(jsonValue.getDate()),jsonValue.getStockCode());
            Boolean insert = exponentialTemperatureDao.insert(temperature);
            if (delete){
                System.out.println("删除成功");
            }else {
                System.out.println("删除失败");
            }
            if (insert){
                System.out.println("新增成功");
            }else {
                System.out.println("新增失败");
            }

//                Pb pb = jsonValue.getPb();
//                PeTtm pe_ttm = jsonValue.getPe_ttm();
//                EchartsEntity echartsEntity = new EchartsEntity();
//                echartsEntity.setStockCode(jsonValue.getStockCode());
//                echartsEntity.setStockName(jsonValue.getStockName());
//                echartsEntity.setDate(jsonValue.getDate());
//                    //pb
//                    echartsEntity.setYear("y10");
//                    echartsEntity.setType("mcw");
//                    echartsEntity.setAvgv(pb.getY10().getMcw().getAvgv());
//                    echartsEntity.setCv(pb.getY10().getMcw().getCv());
//                    echartsEntity.setCvpos(pb.getY10().getMcw().getCvpos());
//                    echartsEntity.setMaxpv(pb.getY10().getMcw().getMaxpv());
//                    echartsEntity.setMaxv(pb.getY10().getMcw().getMaxv());
//                    echartsEntity.setMinv(pb.getY10().getMcw().getMinv());
//                    echartsEntity.setQ2v(pb.getY10().getMcw().getQ2v());
//                    echartsEntity.setQ5v(pb.getY10().getMcw().getQ5v());
//                    echartsEntity.setQ8v(pb.getY10().getMcw().getQ8v());
//                    echartsEntity.setPbPe("pb");
//                   Boolean insert1 = echartsService.insert(echartsEntity);
//                    echartsEntity.setYear("y10");
//                    echartsEntity.setType("mcw");
//                    echartsEntity.setAvgv(pe_ttm.getY10().getMcw().getAvgv());
//                    echartsEntity.setCv(pe_ttm.getY10().getMcw().getCv());
//                    echartsEntity.setCvpos(pe_ttm.getY10().getMcw().getCvpos());
//                    echartsEntity.setMaxpv(pe_ttm.getY10().getMcw().getMaxpv());
//                    echartsEntity.setMaxv(pe_ttm.getY10().getMcw().getMaxv());
//                    echartsEntity.setMinv(pe_ttm.getY10().getMcw().getMinv());
//                    echartsEntity.setQ2v(pe_ttm.getY10().getMcw().getQ2v());
//                    echartsEntity.setQ5v(pe_ttm.getY10().getMcw().getQ5v());
//                    echartsEntity.setQ8v(pe_ttm.getY10().getMcw().getQ8v());
//                    echartsEntity.setPbPe("pe_ttm");
//                Boolean insert2 = echartsService.insert(echartsEntity);
//                   echartsEntity.setYear("y10");
//                   echartsEntity.setType("avg");
//                   echartsEntity.setAvgv(pe_ttm.getY10().getAvg().getAvgv());
//                   echartsEntity.setCv(pe_ttm.getY10().getAvg().getCv());
//                   echartsEntity.setCvpos(pe_ttm.getY10().getAvg().getCvpos());
//                   echartsEntity.setMaxpv(pe_ttm.getY10().getAvg().getMaxpv());
//                   echartsEntity.setMaxv(pe_ttm.getY10().getAvg().getMaxv());
//                   echartsEntity.setMinv(pe_ttm.getY10().getAvg().getMinv());
//                   echartsEntity.setQ2v(pe_ttm.getY10().getAvg().getQ2v());
//                   echartsEntity.setQ5v(pe_ttm.getY10().getAvg().getQ5v());
//                   echartsEntity.setQ8v(pe_ttm.getY10().getAvg().getQ8v());
//                   echartsEntity.setPbPe("pe_ttm");
//                Boolean insert3 = echartsService.insert(echartsEntity);
//                   echartsEntity.setYear("y10");
//                   echartsEntity.setType("avg");
//                   echartsEntity.setAvgv(pe_ttm.getY10().getAvg().getAvgv());
//                   echartsEntity.setCv(pe_ttm.getY10().getAvg().getCv());
//                   echartsEntity.setCvpos(pe_ttm.getY10().getAvg().getCvpos());
//                   echartsEntity.setMaxpv(pe_ttm.getY10().getAvg().getMaxpv());
//                   echartsEntity.setMaxv(pe_ttm.getY10().getAvg().getMaxv());
//                   echartsEntity.setMinv(pe_ttm.getY10().getAvg().getMinv());
//                   echartsEntity.setQ2v(pe_ttm.getY10().getAvg().getQ2v());
//                   echartsEntity.setQ5v(pe_ttm.getY10().getAvg().getQ5v());
//                   echartsEntity.setQ8v(pe_ttm.getY10().getAvg().getQ8v());
//                   echartsEntity.setPbPe("pe_ttm");
//                Boolean insert4 = echartsService.insert(echartsEntity);

//                if (messageBody.length() > 30) {
//                    ExponentialTemperature temperature = JSON.parseObject(messageBody, ExponentialTemperature.class);
//                    Boolean insert = exponentialTemperatureDao.insert(temperature);
//                    if (insert) {
//                        System.out.println("新增成功");
//                    } else {
//                        System.out.println("新增失败");
//                    }
//                } else {
//                    String[] split = messageBody.split("\\+");
//                    Boolean delete = exponentialTemperatureDao.delete(split[0], split[1]);
//                    if (delete) {
//                        System.out.println("删除成功");
//                    } else {
//                        System.out.println("删除失败");
//                    }
//                }




            }
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        } finally {
            if (index < msgs.size()) {
                context.setAckIndex(index + 1);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @PreDestroy
    public void stop() {
        if (consumer != null) {
            consumer.shutdown();
            LOGGER.error("MQ：关闭消费者");
        }
    }
}
